/*
 * Copyright 2023 yifeng Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.yifeng.repo.flink.data.transport.streaming.connectors.mysql;

import static org.apache.flink.util.Preconditions.checkNotNull;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.flink.shaded.netty4.io.netty.util.internal.StringUtil;

import com.yifeng.repo.flink.data.transport.config.DataSourceConfig;
import com.yifeng.repo.flink.data.transport.config.DataSourceConfigFactory;
import com.yifeng.repo.flink.data.transport.config.JdbcSourceOptions;
import com.yifeng.repo.flink.data.transport.dto.DbTypeEnums;
import com.yifeng.repo.flink.data.transport.streaming.functions.FullSourceFunction;

/**
 * @author lijing
 * @since 2023-01-03
 * 构建mysqlSource,返回处理全量同步的YifengFullSourceFunction
 *
 */
public class MySqlSource {

    public static <T> Builder<T> builder() {
        return new Builder<>();
    }

    /**
     * Builder class of {@link MySqlSource}.
     *
     */
    public static class Builder<T> {
    	
        private int port = 3306; // default 3306 port
        private String hostname;
        private String database;
        private String username;
        private String password;
        private String serverTimeZone;
        private String url;
        private String fetchSize;
        private String tableList;
        private Map<String,List<Object>> tableFieldMap;
        //全量读取sql语句
        private Map<String,String> sqlCondition;
        //多表并行的并行度
        private Integer parallelism;
        
        public Builder<T> url(String url) {
            this.url = url;
            return this;
        }
        
        public Builder<T> hostname(String hostname) {
            this.hostname = hostname;
            return this;
        }

        /** Integer port number of the MySQL database server. */
        public Builder<T> port(int port) {
            this.port = port;
            return this;
        }

        /**
         * An optional list of regular expressions that match database names to be monitored; any
         * database name not included in the whitelist will be excluded from monitoring. By default
         * all databases will be monitored.
         */
        public Builder<T> database(String database) {
            this.database = database;
            return this;
        }

        public Builder<T> tableList(String tableList) {
        	this.tableList = tableList;
            return this;
        }
        
        public Builder<T> tableFieldMap(Map<String,List<Object>> tableFieldMap) {
            this.tableFieldMap = tableFieldMap;
            return this;
        }
        
        public Builder<T> sqlCondition(Map<String,String> sqlCondition) {
            this.sqlCondition = sqlCondition;
            return this;
        }
        
        /** Name of the MySQL database to use when connecting to the MySQL database server. */
        public Builder<T> username(String username) {
            this.username = username;
            return this;
        }

        /** Password to use when connecting to the MySQL database server. */
        public Builder<T> password(String password) {
            this.password = password;
            return this;
        }

        /**
         * The session time zone in database server, e.g. "America/Los_Angeles". It controls how the
         * TIMESTAMP type in MYSQL converted to STRING. See more
         * https://debezium.io/documentation/reference/1.5/connectors/mysql.html#mysql-temporal-types
         */
        public Builder<T> serverTimeZone(String timeZone) {
            this.serverTimeZone = timeZone;
            return this;
        }
        
        public Builder<T> fetchSize(String fetchSize) {
            this.fetchSize = fetchSize;
            return this;
        }
        
        public Builder<T> parallelism(Integer parallelism) {
            this.parallelism = parallelism;
            return this;
        }

        public FullSourceFunction<T> build() {
        	DataSourceConfig config = new DataSourceConfigFactory().create();
        	config.setDriverClassName(JdbcSourceOptions.JDBC_DRIVER.defaultValueAsString());
        	config.setDbType(DbTypeEnums.MYSQL.getType());
        	config.setHostname(checkNotNull(hostname));
        	config.setUsername(checkNotNull(username));
        	config.setPassword(checkNotNull(password));
        	config.setPort(port);
        	config.setJdbcUrl(url);
        	config.setDatabase(database);
        	config.setParallelism(parallelism);
        	if (tableList != null) {
        		config.setTableList(Arrays.asList(checkNotNull(tableList).split(",")));
            }
        	if (tableFieldMap != null) {
        		config.setTableFieldMap(tableFieldMap);
            }
        	if (sqlCondition != null) {
        		config.setSqlCondition(sqlCondition);
            }
        	if(!StringUtil.isNullOrEmpty(serverTimeZone)) {
        		config.setServerTimeZone(String.valueOf(serverTimeZone));
        	}
        	if(!StringUtil.isNullOrEmpty(fetchSize)) {
        		config.setFetchSize(Integer.valueOf(fetchSize));
        	}
            return new FullSourceFunction(config, new MySqlValidator());
        }
    }
}
